package com.example.parsing.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.example.parsing.producer.TxtProducer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
public class TxtListener {
	
	private static final Logger logger = LoggerFactory.getLogger(TxtListener.class);
	
	public static Map<String,String> deviceName = new HashMap<>();
	
	@KafkaListener(topicPattern = "sys.ExperimentData.subdevice.register.response")
	public void listen(ConsumerRecord<Object,Object> record){
		logger.info("txt文本收到注册返回的消息"+System.currentTimeMillis()+"收到的消息是"+record.toString());
		try{
			JSONObject registerResponse = (JSONObject) JSON.parse(record.value().toString());
			String id = registerResponse.get("id").toString();
			if(TxtProducer.idMap.containsKey(id)){
				String deviceName2 = TxtProducer.idMap.get(id);
				JSONObject response = (JSONObject) JSON.parse(registerResponse.get("response").toString());
				String status = response.get("status").toString();
				if(status.equals("success")){
					String deviceId = registerResponse.getString("deviceId");
					logger.info("---------------------------txt文件从卡夫卡得到的deviceId="+deviceId);
					deviceName.put(deviceName2,deviceId);
					}
			}
			else {
				logger.info("注册失败可能已经被注册过");
			}
		}catch (Exception e){
			logger.error("未收到消息，错误类型是"+e);
		}
	
	}
}
